package operator;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * connect 算子
 */
public class ConnectDemo1 {

    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> lines1 = env.socketTextStream("localhost", 8888);
        DataStreamSource<String> lines2 = env.socketTextStream("localhost", 9999);
        SingleOutputStreamOperator<Integer> nums = lines2.map(Integer::parseInt);

        // 连接两个Stream
        ConnectedStreams<String, Integer> connect = lines1.connect(nums);

        // ConnectedStream 不能调用sink 需要进行transformation
        SingleOutputStreamOperator<String> maped = connect.map(new CoMapFunction<String, Integer, String>() {

            // 使用状态进行高级操作

            // 对第一个Stream进行处理
            @Override
            public String map1(String s) throws Exception {
                return s;
            }

            // 对第二个Stream进行处理
            @Override
            public String map2(Integer integer) throws Exception {
                return integer.toString();
            }
        });

        maped.print();
        env.execute();
    }
}
